package com.chukun.flink.stream.window.process.windows;

import com.chukun.flink.stream.window.source.SourceForWindow;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author chukun
 * @version 1.0.0
 * @description 窗口 apply函数操作
 * @createTime 2022年05月24日 21:34:00
 */
public class WindowApplyOperator {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Tuple3<String, Integer, String>> streamSource = env.addSource(new SourceForWindow(1000, false));

        SingleOutputStreamOperator<String> applyStream = streamSource.keyBy((key) -> key.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .apply(new WindowFunction<Tuple3<String, Integer, String>, String, String, TimeWindow>() {
                    @Override
                    public void apply(String key, TimeWindow timeWindow, Iterable<Tuple3<String, Integer, String>> input, Collector<String> collector) throws Exception {
                        String str = "";
                        long count = 0;
                        for (Tuple3<String, Integer, String> in : input) {
                            str = StringUtils.join(str, in.toString());
                            count++;
                        }
                        // 将计算的元素个数和窗口信息及窗口的key拼接成一个字符串输出
                        collector.collect("Window: " + timeWindow.toString() + " key: " + key + " count: " + count);
                    }
                });

        applyStream.print("输出结果");

        env.execute("WindowApplyOperator");
    }
}
